package ml

import base.BaseModel
import org.apache.spark.sql.{Column, DataFrame, SparkSession, functions}

object PSM2 extends BaseModel {
  //设置程序名
  override def setAppName: String = "PSM"

  //设置标签id
  override def setFourId: String = "151"

  override def getNewTag(spark: SparkSession, fiveTagDF: DataFrame, HBaseDF: DataFrame): DataFrame = {
    //隐式转化
    import spark.implicits._
    //引入sparkSQL的内置函数
    import org.apache.spark.sql.functions._

    //优惠次数
    var couponCodeSum = functions
      .when('couponCodeValue === 0.0D, 0)
      .when('couponCodeValue =!= 0.0D, 1)
      .as("state")

    //应收总金额 成交价+优惠价格
    var amountReceivable = col("orderAmount") + col("couponCodeValue") as ("amountReceivable")

    val tmpDF: DataFrame = HBaseDF.select('memberId, 'orderSn, 'orderAmount, 'couponCodeValue, couponCodeSum, amountReceivable)

    tmpDF.show()

    //获取优惠次数
    val getCouponCodeSum = functions.sum('state) as ("couponCodeSum")
    //总购买次数
    val getAllSum = functions.count('state) as ("allSum")
    //优惠总金额
    val getCouponCodeAllMoney = functions.sum('couponCodeValue) as("couponCodeAllMoney")
    //应收总金额
    val getReceivableAllMoney = functions.sum('amountReceivable) as ("receivableAllMoney")

    //优惠订单占比
    val getCouponCodePercent = col("couponCodeSum") / col("allSum")
    //平均优惠金额占比
    val getAvgCouponCodeMoneyPercent =
      (col("couponCodeAllMoney") / col("couponCodeSum"))/
        (col("receivableAllMoney")/col("allSum"))
    //优惠总金额占比
    val getCouponCodeAllMoneyPercent = col("couponCodeAllMoney") / col("receivableAllMoney")

    var getPSM=getCouponCodePercent+getCouponCodeAllMoneyPercent+getAvgCouponCodeMoneyPercent as ("PSM")

    tmpDF.groupBy('memberId)
      .agg(getCouponCodeSum,getAllSum,getCouponCodeAllMoney,getReceivableAllMoney)
        .select('memberId,getPSM)
        .filter('PSM isNotNull)
        .show()
    HBaseDF
  }

  //主方法
  def main(args: Array[String]): Unit = {
    exec()
  }
}
